#include "signal_collect.h"

#include "parse_proto.h"
#include "adr_trigger.h"
#include "soaC2SignalToEdc.h"
#include "eventloop.h"
#include "tool.h"
#include "data_build.h"

using namespace DcAdr;

SignalCollect &SignalCollect::getInstance()
{
    static SignalCollect instance;
    return instance;
}

bool SignalCollect::protoRuleContainTopic(const std::string &topic)
{
    return Contain_Map(m_protoParser.m_protoSignalParser, topic);
}

SignalCollect::SignalCollect(/* args */)
{
    m_protoParser.initProto();
    m_protoParser.parseSignalConfigJsonData(global_data::g_collect_config_buf);

    m_protoParser2.initProto();
    m_protoParser2.parseSignalConfigJsonData(global_data::g_collect_config_buf);

    std::vector<std::string> signalnamelist;
    signalnamelist.push_back("BcmPwrStsFb");
    signalnamelist.push_back("AD_RunRedLightWaring");
    signalnamelist.push_back("WarningLevel");
    signalnamelist.push_back("EventNotice");
    signalnamelist.push_back("DriverStatusNeedNotice");
    signalnamelist.push_back("SafeStopSts");
    signalnamelist.push_back("PDStatus");
    signalnamelist.push_back("APAStatus");
    signalnamelist.push_back("PDSubAPAStatus");
    signalnamelist.push_back("RPAStatus");
    signalnamelist.push_back("RADSStatus");
    signalnamelist.push_back("AD_LatTakeOverReq");
    signalnamelist.push_back("AD_LngTakeOverReq");
    g_mqtt_seleced_signal.insert(std::pair<std::string, std::vector<std::string>>("Et_IT064", signalnamelist));
    signalnamelist.clear();
    signalnamelist.push_back("ADS_ADSUseTime");
    signalnamelist.push_back("ADS_AEBTriTime");
    signalnamelist.push_back("ADS_AESTriTime");
    signalnamelist.push_back("ADS_ELKTriTime");
    signalnamelist.push_back("ADS_LaneChangeTime");
    signalnamelist.push_back("ADS_NIDLongestMiles");
    signalnamelist.push_back("ADS_NIDMiles");
    signalnamelist.push_back("ADS_ObsAvoidTime");
    signalnamelist.push_back("ADS_SafeFunTriTime");
    signalnamelist.push_back("ADS_TakeoverTime");
    signalnamelist.push_back("ADS_ThroughIntersectionTime");
    signalnamelist.push_back("ADS_ThroughRampTime");
    signalnamelist.push_back("ADS_TotalMiles");
    signalnamelist.push_back("APA_SaveTime");
    signalnamelist.push_back("APA_UseTime");
    g_mqtt_seleced_signal.insert(std::pair<std::string, std::vector<std::string>>("Et_IT090", signalnamelist));
}

SignalCollect::~SignalCollect()
{
}

//用于给E_SignalCollect 线程做信号解析
void SignalCollect::collectSignalOneThread_1(const SigCollectCatch &data)
{
    if (data.data == nullptr || data.data.get() == nullptr)
    {
        if (data.topic.get() != nullptr)
        {
            Log_Topic("SignalCollect data empty topic: {}", *data.topic.get());
        }
        return;
    }
    uint8_t *buf = (uint8_t *)data.data.get()->c_str();
    if (buf == nullptr)
    {
        if (data.topic.get() != nullptr)
        {
            Log_Signal("SignalCollect data error topic: {}", *data.topic.get());
        }
        return;
    }
    std::string topic = *data.topic.get();
    //获取解析器
    google::protobuf::Message *parser(nullptr);
    parser = m_protoParser.getProtoMessageByTopic(topic);

    if (nullptr == parser)
    {
        Log_Signal("Proto parser not found, topic: {}", topic);
        return;
    }
    if (!parser->ParsePartialFromArray(buf, data._length))
    {
        Log_Signal("Proto parse data failed, topic: {}", topic);
        return;
    }

    std::vector<ProtoCondRuleSignalInfo_t> signalRules;
    signalRules = m_protoParser.m_protoSignalParser[topic];
    // Log_Signal("signalCollect  topic: {}  rulle count: {}", topic, signalRules.size());
    for (ProtoCondRuleSignalInfo_t &signalInfo : signalRules)
    {
        try
        {
            /* code */
             Json::Value value = AdrTrigger::getInstance().ProtoMemberToJsonValue(*parser, signalInfo.name);
            if (value.empty())
            {

                Log_Signal("ERROR: signal get member value failed, signalName: {} ,topic: {} ,pbMsg: {} ,name: {}", signalInfo.signalName, signalInfo.topic, signalInfo.protoMsg, Util::queueToString(signalInfo.name));
                continue;
            }
            m_signalMapValue_1[signalInfo.signalName] = value;
        }
        catch(const std::exception& e)
        {
            Log_Info("gd Exception json value1 ERROR: {}", e.what());
        }
        
       
    }
}

//目前用于给E_SignalCollect_2 线程做信号解析
void SignalCollect::collectSignalOneThread(const SigCollectCatch &data, ThreadType type)
{
    auto funcTri = [this, data, type]()
    {
        if (data.data == nullptr || data.data.get() == nullptr)
        {
            if (data.topic.get() != nullptr)
            {
                Log_Topic("SignalCollect data empty topic: {}", *data.topic.get());
            }
            return;
        }
        uint8_t *buf = (uint8_t *)data.data.get()->c_str();
        if (buf == nullptr)
        {
            if (data.topic.get() != nullptr)
            {
                Log_Signal("SignalCollect data error topic: {}", *data.topic.get());
            }
            return;
        }
        std::string topic = *data.topic.get();
        //获取解析器
        google::protobuf::Message *parser(nullptr);
        if (type == ThreadType::E_SignalCollect)
        {
            parser = m_protoParser.getProtoMessageByTopic(topic);
        }
        else if (type == ThreadType::E_SignalCollect_2)
        {
            parser = m_protoParser2.getProtoMessageByTopic(topic);
        }

        if (nullptr == parser)
        {
            Log_Signal("Proto parser not found, topic: {}", topic);
            return;
        }
        if (!parser->ParsePartialFromArray(buf, data._length))
        {
            Log_Signal("Proto parse data failed, topic: {}", topic);
            return;
        }

        std::vector<ProtoCondRuleSignalInfo_t> signalRules;
        if (type == ThreadType::E_SignalCollect_2)
        {
            signalRules = m_protoParser2.m_protoSignalParser[topic];
        }
        else if (type == ThreadType::E_SignalCollect)
        {
            signalRules = m_protoParser.m_protoSignalParser[topic];
        }
        // Log_Signal("signalCollect  topic: {}  rulle count: {}", topic, signalRules.size());
        for (ProtoCondRuleSignalInfo_t &signalInfo : signalRules)
        {
            try
            {
                /* code */
                 Json::Value value = AdrTrigger::getInstance().ProtoMemberToJsonValue(*parser, signalInfo.name);
                if (value.empty())
                {

                    Log_Signal("ERROR: signal get member value failed, signalName: {} ,topic: {} ,pbMsg: {} ,name: {}", signalInfo.signalName, signalInfo.topic, signalInfo.protoMsg, Util::queueToString(signalInfo.name));
                    continue;
                }
                m_signalMapValue_2[signalInfo.signalName] = value;

            }
            catch(const std::exception& e)
            {
                Log_Info("gd Exception json value2 ERROR: {}", e.what());
            }
            
           
        }
    };
    appendEvent(type, funcTri);
}

//目前用于给E_SignalCollect_2 线程处理m_signalMapValue_2
void SignalCollect::clearSignaledMap()
{
    auto func = [=]()
    {
        m_signalMapValue_2.clear();
    };
    appendEvent(ThreadType::E_SignalCollect_2, func);
}

//目前用于给E_SignalCollect_2 线程处理交换m_signalMapValue_2里解析的信号到周期/变化map
void SignalCollect::swapSignaledMap()
{
    auto func = [=]()
    {
        m_dataLock.lock();
        for (auto iter : m_signalMapValue_2)
        {
            //以cyc为缓存老的，新到的不同放入change
            if (NotContain_Map(m_cycSignalData, iter.first))
            {
                m_changeSignalData[iter.first] = iter.second;
            }
            else
            {
                if (m_cycSignalData[iter.first] != iter.second)
                {
                    m_changeSignalData[iter.first] = iter.second;
                }
            }

            m_cycSignalData[iter.first] = iter.second;
        }
        m_dataLock.unlock();
    };
    appendEvent(ThreadType::E_SignalCollect_2, func);
}

void SignalCollect::push2SigMap(const ProtoData &data)
{
    auto func = [=]()
    {
        SigCollectCatch sigData = {data.topic, data.data, data._length};
        m_signalCache.insert({*data.topic.get(), sigData});
    };
    appendEvent(ThreadType::E_SignalCollect, func);
    
}

void SignalCollect::collectSignal()
{
    auto func = [=]()
    {
        m_signalMapValue_1.clear();
        clearSignaledMap();
        static int i = 0;

        for (auto beg = m_signalCache.begin(); beg != m_signalCache.end(); ++beg)
        {
            SigCollectCatch &data = beg->second;
            if (i % 2 == 0)
            {
                collectSignalOneThread_1(data);
            }
            else
            {
                collectSignalOneThread(data, ThreadType::E_SignalCollect_2);
            }
            i++;

            if (i == 10)
            {
                i = 0;
            }
        }
        m_signalCache.clear();
        swapSignaledMap();
        {
            m_dataLock.lock();
            for (auto iter : m_signalMapValue_1)
            {
                //以cyc为缓存老的，新到的不同放入change
                if (NotContain_Map(m_cycSignalData, iter.first))
                {
                    m_changeSignalData[iter.first] = iter.second;
                }
                else
                {
                    if (m_cycSignalData[iter.first] != iter.second)
                    {
                        m_changeSignalData[iter.first] = iter.second;
                    }
                }

                m_cycSignalData[iter.first] = iter.second;                
            }
            m_dataLock.unlock();
        }
    };
    appendEvent(ThreadType::E_SignalCollect, func);
}

void SignalCollect::sendSignalToMqtt()
{
    for(auto iter : g_mqtt_seleced_signal)
    {
        try
        {
            /* code */
            Json::Value params;
            for(auto iter1: iter.second)
            {
                // Log_Info("gd mqtt collect signal mk {} signalname {}",iter.first,iter1);
                if(Contain_Map(m_changeSignalData, iter1))
                {
                    params[iter1] = m_changeSignalData[iter1];
                    m_changeSignalData.erase(iter1);
                }
            }
            if(params.size() > 0)
            {
                    DataBuild::DataCollectSignal_send("Et_CarData", iter.first, params);
            }
        }
        catch(const std::exception& e)
        {
            Log_Info("gd Exception occurred: {}", e.what());
        }
        
        
    }
}

void SignalCollect::sendSignalShot()
{  

    SignalCollect::getInstance().collectSignal();//采集完变化的直接加入send线程队列发

    //发送change的
    auto func = [this]()
    {
        try
        {
            /* code */
            if (m_changeSignalData.empty())
            {
                return;
            }
            Json::Value value;
            m_dataLock.lock();
            sendSignalToMqtt();
            for (auto &signal : m_changeSignalData)
            {
                value[signal.first] = signal.second;
            }
            m_changeSignalData.clear();
            m_dataLock.unlock();
            // test(value);//zjs_todelet_4;
            Json::FastWriter writer;
            // 去DDS
            SoaC2SignalToEdc::getInstance().sendSignal(writer.write(value));
        }
        catch(const std::exception& e)
        {
            Log_Info("gd Exception occurred1: {}", e.what());
        }
              
    };
    appendEvent(ThreadType::E_SendSignal, func);

    if (m_TimerCount == 3000)
    {
        //发送cyc的
        auto func = [this]()
        {

            try
            {
                /* code */
                if (m_cycSignalData.empty())
                {
                    return;
                }
                Json::Value value;
                m_dataLock.lock();
                for (auto &signal : m_cycSignalData)
                {
                    value[signal.first] = signal.second;
                }
                m_dataLock.unlock();
                // test(value);//zjs_todelet_4;
                Json::FastWriter writer;
                // 去DDS
                SoaC2SignalToEdc::getInstance().sendSignal(writer.write(value));
                // DdsC2SignalToEdc::getInstance().sendSignal(writer.write(value));
            }
            catch(const std::exception& e)
            {
                Log_Info("gd Exception occurred2: {}", e.what());
            }
            
        };
        appendEvent(ThreadType::E_SendSignal, func);

        m_TimerCount = 0;
    }
    m_TimerCount += 200;
}

void SignalCollect::test(Json::Value &root)
{
    static int i = 0;
    Json::Value::Members member = root.getMemberNames();
    for (Json::Value::Members::iterator iter = member.begin(); iter != member.end(); ++iter)
    {
        i++;
        std::string memberName = *iter;
        root[memberName] = i;
    }
    if (i > 200000)
    {
        i = 0;
    }
}